package com.suxin.datahub.core.writer;

import com.suxin.datahub.database.DataBaseFactory;
import com.suxin.datahub.domain.DataSourceDomain;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * @Classname AbstractReader
 * @Description [ AbstractReader ]
 * @Author Tang
 * @Date 2020/11/11 13:12
 * @Created by ASUS
 */
@Slf4j
public abstract class AbstractBaseWriter implements Writer {

    public Connection connection = null;

    private DataSourceDomain dataSourceDomain;

    public AbstractBaseWriter(DataSourceDomain dataSourceDomain) {

        this.connection = DataBaseFactory.getDataSource(dataSourceDomain);

        this.dataSourceDomain = dataSourceDomain;
    }

    @Override
    public long batchWriterData(List<Map<String, Object>> readData) {
        // 获取sql
        String[] columns = readData.get(0).keySet().toArray(new String[]{});
        String sql = this.dataSourceDomain.getSql() + Arrays.toString(columns)
                .replace("[", "(")
                .replace("]", ")");
        StringBuilder valueBuilder = new StringBuilder(" value (");
        for (int i = 0; i < columns.length; i++) {
            valueBuilder.append("?,");
        }
        valueBuilder.deleteCharAt(valueBuilder.length() - 1);
        valueBuilder.append(")");
        sql = sql + valueBuilder;
        log.info("sql范式:" + sql);
        // 批量插入
        long row = 0;
        try {
            connection.setAutoCommit(false);
            PreparedStatement preparedStatement = connection.prepareStatement(sql);
            int index = 1;
            for (Map<String, Object> item : readData) {
                for (String key : item.keySet()) {
                    preparedStatement.setObject(index++,item.get(key));
                }
                index = 1;
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
            connection.commit();
            log.info("执行成功，提交事务成功...");
            return row;
        } catch (SQLException throwables) {
            try {
                log.error("执行错误，开始事务回滚...");
                connection.rollback();
            } catch (SQLException throwable) {
                throwable.printStackTrace();
            }
        }
        return 0;
    }
}